package com.bujunjie.study.realtime.dws.app;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.bujunjie.study.realtime.common.function.Object2JsonStringFunction;
import com.bujunjie.study.realtime.dws.app.model.sink.CartAddUserModel;
import com.bujunjie.study.realtime.dws.app.model.vo.CartModelInfoVO;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Date;

/**
 * <p>dws 购物车添加交易</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/17 17:05
 */
public class DwsTradeCartAddUuWindow extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwsTradeCartAddUuWindow().start(10026, 4, "dws_trade_cart_add_uu_window", FlinkConstant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {

        // 购物车数据流
        SingleOutputStreamOperator<CartModelInfoVO> cartDS = kafkaStrDS.map(x -> JSON.parseObject(x, new TypeReference<CartModelInfoVO>() {
        }));
        // 生成水位线的数据流
        SingleOutputStreamOperator<CartModelInfoVO> withWatermarkDS = cartDS.assignTimestampsAndWatermarks(WatermarkStrategy.<CartModelInfoVO>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<CartModelInfoVO>() {
            @Override
            public long extractTimestamp(CartModelInfoVO cartModelInfoVO, long recordTimestamp) {
                return cartModelInfoVO.getTs() * 1000;
            }
        }));
        // 按照键进行分区处理
        KeyedStream<CartModelInfoVO, String> keyByDS = withWatermarkDS.keyBy(CartModelInfoVO::getUserId);

        //  过滤出有问题的数据流
        SingleOutputStreamOperator<CartModelInfoVO> cartUserIdDS = keyByDS.process(new KeyedProcessFunction<String, CartModelInfoVO, CartModelInfoVO>() {
            private ValueState<Long> timeState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> timeStateDesc = new ValueStateDescriptor<>("timeStateDesc", Long.class);
                timeStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                timeState = getRuntimeContext().getState(timeStateDesc);
            }

            @Override
            public void processElement(CartModelInfoVO cartModel, KeyedProcessFunction<String, CartModelInfoVO, CartModelInfoVO>.Context ctx, Collector<CartModelInfoVO> out) throws Exception {
                Long ts = cartModel.getTs();
                if (ObjectUtil.isNotNull(ts) || !timeState.value().equals(ts * 1000)) {
                    timeState.update(ts * 1000);
                    out.collect(cartModel);
                }
            }
        });

        AllWindowedStream<CartModelInfoVO, TimeWindow> windowDS = cartUserIdDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        SingleOutputStreamOperator<CartAddUserModel> aggDS = windowDS.aggregate(new AggregateFunction<CartModelInfoVO, Long, Long>() {
            @Override
            public Long createAccumulator() {
                return 0L;
            }


            @Override
            public Long add(CartModelInfoVO value, Long accumulator) {
                return ++accumulator;
            }

            @Override
            public Long getResult(Long accumulator) {
                return accumulator;
            }

            @Override
            public Long merge(Long a, Long b) {
                return 0L;
            }
        }, new AllWindowFunction<Long, CartAddUserModel, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<Long> values, Collector<CartAddUserModel> out) throws Exception {
                String start = DateUtil.format(new Date(window.getStart()), DatePattern.NORM_DATETIME_PATTERN);
                String end = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATETIME_PATTERN);
                String currentDate = DateUtil.format(new Date(window.getEnd()), DatePattern.NORM_DATE_PATTERN);
                CartAddUserModel userModel = CartAddUserModel.builder().startTime(start).endTime(end).currentDate(currentDate).cartAddUuCnt(values.iterator().next()).build();
                out.collect(userModel);
            }
        });

        aggDS
                //  数据转换
                .map(new Object2JsonStringFunction<CartAddUserModel>())
                //  将数据写入 doris 中
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_cart_add_uu_window"));


    }
}
